package com.ddl.kafka.controller;

import com.ddl.kafka.config.direct.CountProducer;

import com.ddl.kafka.config.topic.TopicProducer;
import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@Slf4j
public class MsgController {

    @Autowired
    private CountProducer countProducer;

    @Autowired
    private TopicProducer topicProducer;

    @GetMapping("/sendMsg2ArtisanQueue")
    public void sendMsg() throws InterruptedException {
        for(int i = 0; i< 100; i++) {

            countProducer.asyncSend("sendMsg" + i).addCallback(new ListenableFutureCallback<Void>() {

                @Override
                public void onFailure(Throwable e) {
                    log.info("[testASyncSend][发送编号：[{}] 发送异常]]", e);
                }

                @Override
                public void onSuccess(Void aVoid) {
                    log.info("[testASyncSend][发送编号：[{}] 发送成功]");
                }

            });
        }

        // 阻塞等待，保证消费
        //new CountDownLatch(1).await();
    }

    @GetMapping("/sendTopicMsg")
    public void sendTopicMsg() {
        int id = (int) (System.currentTimeMillis() / 1000);
        topicProducer.syncSend(id, "abc.send.email");
    }
    /**
     * 批量发送消息
     */
    /*@GetMapping("batchSendMsg")
    public void batchSendMsg() throws Exception {
        for (int i = 0; i < 3; i++) {
            // 同步发送消息
            int id = (int) (System.currentTimeMillis() / 1000);
            countProducer.batchSyncSend(id, "batch send" + i);

            // 故意每条消息之间，隔离 10 秒
            log.info("[testSyncSend][发送编号：[{}] 发送成功]", id);
            Thread.sleep(10 * 1000L);
        }
    }*/
}
